package io.kiki.sba.registry.server.clientInterface.scheduler.timertask;

import io.kiki.sba.registry.common.model.store.BaseInfo;
import io.kiki.sba.registry.common.model.store.Publisher;
import io.kiki.sba.registry.common.model.store.Subscriber;
import io.kiki.sba.registry.common.model.store.Url;
import io.kiki.sba.registry.server.clientInterface.bootstrap.Config;
import io.kiki.sba.registry.server.clientInterface.store.Publishers;
import io.kiki.sba.registry.server.clientInterface.store.Subscribers;
import io.kiki.sba.registry.util.ConcurrentUtils;
import io.kiki.sba.registry.util.NamedThreadFactory;
import org.apache.commons.lang.time.DateUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;


public class SessionCacheDigestTask {

    private static final Logger logger = LoggerFactory.getLogger("CACHE-DIGEST");
    private final ScheduledExecutorService scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("CacheDigestTask"));
    @Autowired
    Publishers sessionPublishers;
    @Autowired
    Subscribers sessionSubscribers;
    @Autowired
    Config config;

    @PostConstruct
    public boolean init() {
        final int configCacheDigestIntervalMinutes = config.getCacheDigestIntervalMinutes();
        if (configCacheDigestIntervalMinutes <= 0) {
            logger.info("cache digest off with intervalMinutes={}", configCacheDigestIntervalMinutes);
            return false;
        }
        Date firstDate = new Date();
        firstDate = DateUtils.round(firstDate, Calendar.MINUTE);
        firstDate.setMinutes(firstDate.getMinutes() / configCacheDigestIntervalMinutes * configCacheDigestIntervalMinutes + configCacheDigestIntervalMinutes);
        long firstDelay = firstDate.getTime() - System.currentTimeMillis();
        scheduledThreadPoolExecutor.scheduleAtFixedRate(this::dump, firstDelay, (long) configCacheDigestIntervalMinutes * 60 * 1000, TimeUnit.MILLISECONDS);

        return true;
    }

    boolean dump() {
        try {
            Collection<String> storeDataInfoIds = sessionPublishers.getDataInfoIds();
            Collection<String> interestDataInfoIds = sessionSubscribers.getDataInfoIds();
            Set<String> dataInfoIds = new HashSet<>(storeDataInfoIds.size() + interestDataInfoIds.size());

            dataInfoIds.addAll(storeDataInfoIds);
            dataInfoIds.addAll(interestDataInfoIds);

            for (String dataInfoId : dataInfoIds) {
                Collection<Publisher> publishers = sessionPublishers.getDataList(dataInfoId);
                Collection<Subscriber> subscribers = sessionSubscribers.getDataList(dataInfoId);

                logger.info("[dataInfo] {}; {}; {}; {}; [{}]; [{}]", config.getDataCenter(), dataInfoId, publishers.size(), subscribers.size(), logPubOrSub(publishers), logPubOrSub(subscribers));
                // avoid io is too busy
                ConcurrentUtils.sleepUninterruptibly(1, TimeUnit.MILLISECONDS);
            }
            return true;
        } catch (Throwable t) {
            logger.error("[CacheDigestTask] cache digest error", (Throwable) t);
            return false;
        }
    }

    private String logPubOrSub(Collection<? extends BaseInfo> infos) {

        return Optional.ofNullable(infos).orElse(new ArrayList<>()).stream().filter(info -> info != null).map(info -> logUrl(info.getClientUrl())).collect(Collectors.joining(","));
    }

    private String logUrl(Url url) {
        return url == null ? "null" : url.buildAddressString();
    }
}
